#include <string>
#include <vector>
#include <iostream>
using namespace std;
#include "sys/thread.h"
#include "mem/rblock_queue.h"
#include "mem/small_buffer.h"
#include "component/logger.h"
using namespace tdl;

/////////////////
struct uinfo
{
	typedef small_buffer<64>		buffer_type;
	typedef buffer_type::size_type	size_type;
public:
	uint32_t	uin;
	uint32_t	cip;
	buffer_type	data;
public:
	uinfo() : uin(0), cip(0) {}
	
	uinfo(uint32_t u, uint32_t i) : uin(u), cip(i) {}
	
	size_t length() const
	{ return sizeof(uint32_t) * 2 + sizeof(size_t); }

	void serialize(char* p) const
	{
		*(uint32_t*)p = uin;		p += sizeof(uint32_t);
		*(uint32_t*)p = cip;		p += sizeof(uint32_t);
		*(size_type*)p = data.len;	p += sizeof(size_type);
		memcpy(p, data.ptr, data.len);
	}
	void deserialize(const char* p)
	{
		uin = *(uint32_t*)p;		p += sizeof(uint32_t);
		cip = *(uint32_t*)p;		p += sizeof(uint32_t);
		data.len = *(size_type*)p;	p += sizeof(size_type);
		memcpy(data.ptr, p, data.len);
	}
	friend inline ostream& operator<<(ostream& os, const uinfo& ui)
	{
		os << ui.uin << "|" << ui.cip << "|" << *(int*)(ui.data.ptr);
		return os;
	}
};

console_logger log;
rblock_queue_ex<> rbq;

void writer()
{
	int i = 0;
	while(1)
	{
		uinfo ui(i, i);
		ui.data.len = sizeof(uint32_t);
		*(int*)ui.data.ptr = i;
		if(rbq.push(ui, ui.length()))
			++i;
		else
			log.info() << "full" << endl;
		usleep(10000);
	}
}

void reader()
{
	uinfo ui;
	pthread_t pid = ::pthread_self();

	while(1)
	{
		size_t rl;
		rbq.pop(ui, rl);	// will block here
		log.info() << pid << ui << endl;
		usleep(100000);
	}
}

//////////////////////
int main(int argc, char* argv[])
{
	rbq.initialize(1000);
	log.initialize();

	thread wr_thr(make_fun_exec(&writer), false);
	thread_pool tp(5, make_fun_exec(&reader));

	tp.run();
	wr_thr.run();
	wr_thr.wait();

	return 0;
}

